package org.apache.lucene.ocean.util;

import java.util.Iterator;
import java.util.LinkedList;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Abstract class that pools array objects.  
 * @param <T> Type of object being pooled
 */
public abstract class ObjectArrayPool<T> {
  final static Logger LOG = LoggerFactory.getLogger(ObjectArrayPool.class);
  public static final long DEFAULT_EVICT_TIME = 1000 * 60;
  /**
   * Size of the array to the ObjectArrays map
   */
  private final ConcurrentHashMap<Integer,ObjectArrays> map = new ConcurrentHashMap<Integer,ObjectArrays>();
  /**
   * Single threaded timer for the eviction process
   */
  private final ScheduledExecutorService timer;
  /**
   * Duration the object has not been in use before it is evicted
   */
  private final long evictTime;
  
  /**
   * 
   * @param evictTime Time to elapse before an object is removed
   */
  public ObjectArrayPool(long evictTime) {
    this.evictTime = evictTime;
    timer = Executors.newSingleThreadScheduledExecutor();
    timer.scheduleWithFixedDelay(new Timer(), evictTime, evictTime, TimeUnit.MILLISECONDS);
  }
  
  /**
   * Timer eviction runnable
   */
  public class Timer implements Runnable {
    public void run() {
      for (ObjectArrays objectArrays : map.values()) {
        objectArrays.evict();
      }
    }
  }

  public static class Entry<T> {
    public final T obj;
    public long accessTime;

    public Entry(T obj) {
      this.obj = obj;
    }
  }
  
  /**
   * Contains a pool of object arrays of a specific size
   *
   */
  public class ObjectArrays {
    public final int length;
    private LinkedList<Entry<T>> queue = new LinkedList<Entry<T>>();
    private ReentrantLock lock = new ReentrantLock();

    public ObjectArrays(int length) {
      this.length = length;
    }

    public void evict() {
      lock.lock();
      try {
        long now = Timeout.TIMER_THREAD.getTime();
        Iterator<Entry<T>> iterator = queue.iterator();
        while (iterator.hasNext()) {
          Entry<T> entry = iterator.next();
          if ((now - entry.accessTime) > evictTime) {
            iterator.remove();
            LOG.info("removed entry length: " + length);
          }
        }
      } finally {
        lock.lock();
      }
    }

    public Entry<T> add(T obj) {
      lock.lock();
      try {
        Entry<T> entry = new Entry<T>(obj);
        queue.add(entry);
        return entry;
      } finally {
        lock.lock();
      }
    }

    public T get() {
      lock.lock();
      try {
        Entry<T> entry = null;
        if (queue.size() > 0) {
          entry = queue.removeLast();
        }
        if (entry == null) {
          return newObject(length);
        }
        entry.accessTime = Timeout.TIMER_THREAD.getTime();
        return entry.obj;
      } finally {
        lock.lock();
      }
    }
  }
  
  /**
   * Method that needs to be overridden by subclasses that instantiates a new 
   * array of the given length.
   * @param length
   * @return
   */
  protected abstract T newObject(int length);
  
  /**
   * Returns an array object of the given length
   * @param length
   * @return
   */
  public T getObject(int length) {
    ObjectArrays objectArrays = getObjectArrays(length);
    return objectArrays.get();
  }

  protected ObjectArrays getObjectArrays(int length) {
    ObjectArrays objectArrays = map.get(length);
    if (objectArrays == null) {
      objectArrays = new ObjectArrays(length);
      ObjectArrays absentObjectArrays = map.putIfAbsent(length, objectArrays);
      if (absentObjectArrays != null)
        objectArrays = absentObjectArrays;
    }
    return objectArrays;
  }
  
  /**
   * Method that needs to be overridden by subclasses to accept an array
   * object that is no longer being used.
   * @param obj
   */
  public abstract void returnObject(T obj);
}